package com.atguigu.upp.app;

import com.atguigu.upp.bean.TagInfo;
import com.atguigu.upp.service.CKDBService;
import com.atguigu.upp.service.MysqlDBService;
import com.atguigu.upp.utils.DBUtil;
import com.atguigu.upp.utils.PropertiesUtil;
import com.atguigu.upp.utils.TypeCodeConstant;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2022/10/17
 */
public class ExecuteBitmapTask
{
    public static void main(String[] args) throws IOException {

        Integer taskId = Integer.valueOf(args[0]);
        String doDate = args[1];

        SqlSessionFactory sqlSessionFactory1 = DBUtil.getSqlSessionFactory("mybatis_ck.xml");
        SqlSession sqlSession1 = sqlSessionFactory1.openSession();
        CKDBService ckDBService = new CKDBService(sqlSession1);
        SqlSessionFactory sqlSessionFactory2 = DBUtil.getSqlSessionFactory("mybatis_mysql.xml");
        SqlSession sqlSession2 = sqlSessionFactory2.openSession();
        MysqlDBService mysqlDBService = new MysqlDBService(sqlSession2);
        //查询今天要计算的标签有哪些
        List<TagInfo> tags = mysqlDBService.getTagInfoNameEnableToday();

        //根据TagInfo的tagValueType将所有的标签按照值的类型分为4类，分别写入4张bitmap表中
        List<TagInfo> stringTags = new ArrayList<>();
        List<TagInfo> bigintTags = new ArrayList<>();
        List<TagInfo> dateTags = new ArrayList<>();
        List<TagInfo> decimalTags = new ArrayList<>();

        //将标签分类到集合中
        for (TagInfo tag : tags) {

            switch (tag.getTagValueType()){

                case TypeCodeConstant.TAG_VALUE_TYPE_LONG :bigintTags.add(tag); break;
                case TypeCodeConstant.TAG_VALUE_TYPE_DECIMAL : decimalTags.add(tag); break;
                case TypeCodeConstant.TAG_VALUE_TYPE_STRING : stringTags.add(tag); break;
                case TypeCodeConstant.TAG_VALUE_TYPE_DATE : dateTags.add(tag); break;
            }

        }

        //处理集合中的标签，每个集合将所有的标签计算成bitmap分群，写入对应的bitmap表
        insertBitmapTable(stringTags,"user_tag_value_string",ckDBService,doDate);
        insertBitmapTable(bigintTags,"user_tag_value_long",ckDBService,doDate);
        insertBitmapTable(dateTags,"user_tag_value_date",ckDBService,doDate);
        insertBitmapTable(decimalTags,"user_tag_value_decimal",ckDBService,doDate);


        //任务在计算时，是统一使用Spark_submit提交到yarn，一个SparkApp必须有SparkContext，否则就不是一个SparkApp
        SparkSession sparkSession = DBUtil.getSparkSession("ExecuteBitmapTask");


    }

    private static void insertBitmapTable(List<TagInfo> tags,String table,CKDBService ckDBService,String doDate) {

        if (tags.size() > 0){

            //先删除之前已经写过的数据，保证幂等性
            ckDBService.deleteOldPartitionData(table,doDate);


            //获取当天ck中的宽表名
            String sourceTable = PropertiesUtil.getProperty("upwideprefix") + doDate.replace('-','_');  //表名

            /*
                    生成今天要计算的标签列的信息

                    ('gender', gender),
                    ('agegroup', agegroup),
                    ('favor', favor)
             */
            List<String> tagCodes = tags
                .stream()
                .map(tagInfo -> String.format(" ('%s', %s)  ", tagInfo.getTagCode().toLowerCase(), tagInfo.getTagCode().toLowerCase()))
                .collect(Collectors.toList());

            String columnStr = StringUtils.join(tagCodes, ",");

            //再执行写入
            ckDBService.insertBitmap(table,columnStr,sourceTable,doDate);

        }

    }
}
